package cn.llyong.rpc.client.handler;

import cn.llyong.rpc.common.bo.RpcRequest;
import cn.llyong.rpc.common.bo.RpcResponse;
import com.baidu.bjf.remoting.protobuf.Codec;
import com.baidu.bjf.remoting.protobuf.ProtobufProxy;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.SynchronousQueue;

/**
 * @description:
 * @author: lvyong
 * @date: 2020年06月30日 08:52
 */
public class RpcClientHandler extends ChannelDuplexHandler {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static Map<String, SynchronousQueue<Object>> queueMap = new ConcurrentHashMap<>();

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        ctx.writeAndFlush(Unpooled.copiedBuffer( (byte[]) msg));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx){

    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg){
        Codec<RpcResponse> codec = ProtobufProxy.create(RpcResponse.class);
        try {
            ByteBuf buf = (ByteBuf)msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            RpcResponse response = codec.decode(req);
            logger.info("服务端响应对象：{}", response);
            SynchronousQueue<Object> queue = queueMap.get(response.getRequestId());
            queue.put(response);
            queueMap.remove(response.getRequestId());
        } catch (Exception e) {
            logger.error("rpc 调用异常！！！", e);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("出错了！！！", cause);
        ctx.close();
    }

    public SynchronousQueue<Object> send(RpcRequest request, Channel channel) throws Exception {
        SynchronousQueue<Object> queue = new SynchronousQueue<>();
        queueMap.put(request.getRequestId(), queue);

        Codec<RpcRequest> codec = ProtobufProxy.create(RpcRequest.class);
        byte[] encode = codec.encode(request);

        channel.writeAndFlush(encode);
        return queue;
    }
}
